package com.duoec.dw.lineage.dto;

import com.duoec.dw.lineage.enums.DataDirectionEnum;
import com.duoec.dw.lineage.enums.StatementType;

import java.util.LinkedHashSet;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * 某个表的血缘关系（上游依赖）
 *
 * @author xuwenzhen
 */
public class TableLineageDto extends TableCoordinateDto {
    public static final String STR_TURN_LINE_T2 = "\n\t\t";
    public static final String STR_TURN_LINE_T3 = "\n\t\t\t";

    /**
     * 表别名
     */
    private String alias;

    /**
     * 字段列表
     */
    private LinkedHashSet<TableFieldLineageDto> fields;

    /**
     * 表说明
     */
    private String comment;

    /**
     * 整表依赖关系
     */
    private LinkedHashSet<TableFieldDependencyDto> dependencies;

    /**
     * Partitions
     */
    private List<String> partitions;

    /**
     * 是否来自 lateral view
     */
    private Boolean lateralView;

    /**
     * 语句类型
     */
    private StatementType statementType;

    /**
     * 表类型：0=普通表 1=临时表(未删除) 2=临时表(已删除)
     */
    private Integer tableType = 0;

    /**
     * 脚本信息
     */
    private SqlScriptDto script;

    /**
     * 事件类型，十位数表示来源库类型，个位数表示去
     * 0表示hive表 1表示mysql 2表示mongodb
     * 0 = hive -> hive
     * 1 = hive -> mysql
     * 2 = hive -> mongodb
     * 10= mysql -> hive
     * @see DataDirectionEnum
     */
    private Integer dataDirection;

    public LinkedHashSet<TableFieldLineageDto> getFields() {
        return fields;
    }

    public void setFields(LinkedHashSet<TableFieldLineageDto> fields) {
        this.fields = fields;
    }

    public String getComment() {
        return comment;
    }

    public void setComment(String comment) {
        this.comment = comment;
    }

    public String getAlias() {
        return alias;
    }

    public void setAlias(String alias) {
        this.alias = alias;
    }

    public LinkedHashSet<TableFieldDependencyDto> getDependencies() {
        return dependencies;
    }

    public void setDependencies(LinkedHashSet<TableFieldDependencyDto> dependencies) {
        this.dependencies = dependencies;
    }

    public List<String> getPartitions() {
        return partitions;
    }

    public void setPartitions(List<String> partitions) {
        this.partitions = partitions;
    }

    public void setLateralView(Boolean lateralView) {
        this.lateralView = lateralView;
    }

    public StatementType getStatementType() {
        return statementType;
    }

    public void setStatementType(StatementType statementType) {
        this.statementType = statementType;
    }

    public Integer getTableType() {
        return tableType;
    }

    public void setTableType(Integer tableType) {
        this.tableType = tableType;
    }

    public SqlScriptDto getScript() {
        return script;
    }

    public void setScript(SqlScriptDto script) {
        this.script = script;
    }

    public Integer getDataDirection() {
        return dataDirection;
    }

    public void setDataDirection(Integer dataDirection) {
        this.dataDirection = dataDirection;
    }

    public TableFieldLineageDto getFieldByName(String fieldName) {
        if (fields == null) {
            return null;
        }
        Optional<TableFieldLineageDto> firstOpt = fields.stream()
                .filter(field -> fieldName.equals(field.getColName()))
                .findFirst();
        return firstOpt.orElse(null);
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder(getTableFullName(this));

        if (statementType != null && statementType != StatementType.INSERT_TABLE_DATA) {
            sb.append(" [").append(statementType.name()).append("]");
            if (fields == null) {
                return sb.toString();
            }
        }

        if (dependencies != null && !dependencies.isEmpty()) {
            sb.append("\n\tglobalDependencies:");
            getDependencyString(sb, dependencies, STR_TURN_LINE_T2);
        }
        if (fields != null && !fields.isEmpty()) {
            sb.append("\n\tfields:");
            int[] index = new int[]{0};
            fields.forEach(field -> {
                sb.append(STR_TURN_LINE_T2).append(index[0]).append(".\t").append(field.getColName()).append(":");
                index[0] += 1;
                getDependencyString(sb, field.getDependencies(), STR_TURN_LINE_T3);
            });
        }
        if (partitions != null && !partitions.isEmpty()) {
            sb.append("\n\tpartitions:");
            partitions.forEach(item -> sb.append(STR_TURN_LINE_T2).append(item));
        }
        return sb.toString();
    }

    private void getDependencyString(StringBuilder sb, LinkedHashSet<TableFieldDependencyDto> dependencies, String s) {
        if (dependencies == null) {
            return;
        }
        int[] index = new int[]{0};
        dependencies.forEach(dependency -> {
            sb.append(s)
                    .append(index[0])
                    .append(".\t")
                    .append(getTableFullName(dependency))
                    .append(".")
                    .append(dependency.getFieldName());
            index[0] += 1;
        });
    }

    public String getFullTableName() {
        return super.toString();
    }

    private String getTableFullName(TableCoordinateDto tableCoordinate) {
        StringBuilder sb = new StringBuilder();
        if (tableCoordinate.getDatabaseName() != null) {
            sb.append(tableCoordinate.getDatabaseName());
        }
        if (tableCoordinate.getTableName() != null) {
            if (sb.length() > 0) {
                sb.append(".");
            }
            sb.append(tableCoordinate.getTableName());
        }
        return sb.toString();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof TableLineageDto)) {
            return false;
        }
        if (!super.equals(o)) {
            return false;
        }
        TableLineageDto that = (TableLineageDto) o;
        return Objects.equals(databaseName, that.databaseName) &&
                tableName.equals(that.tableName) &&
                Objects.equals(alias, that.alias);
    }

    @Override
    public int hashCode() {
        return Objects.hash(super.hashCode(), databaseName, tableName, alias);
    }

    public Boolean getLateralView() {
        return lateralView;
    }

    public TableFieldLineageDto getField(String fieldName) {
        if (fields == null || fieldName == null) {
            return null;
        }
        return fields.stream()
                .filter(field -> fieldName.equals(field.getColName()))
                .findFirst()
                .orElse(null);
    }
}
